/*
 * Copyright (c) 2023 Huawei Device Co., Ltd.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Description: Cast Session implement realization.
 * Author: lijianzhao
 * Create: 2022-01-25
 */

#include "cast_session_impl.h"

#include <array>

#include "cast_engine_errors.h"
#include "cast_engine_log.h"
#include "cast_device_data_manager.h"
#include "connection_manager.h"
#include "ipc_skeleton.h"
#include "json/json.h"
#include "permission.h"
#include "dlna_dmr.h"

namespace OHOS {
namespace CastEngine {
namespace CastEngineService {
DEFINE_CAST_ENGINE_LABEL("Cast-SessionImpl");

using CastSessionRtsp::ActionType;
using CastSessionRtsp::DeviceTypeParamInfo;

const std::array<std::string, static_cast<size_t>(CastSessionImpl::MessageId::MSG_ID_MAX)>
    CastSessionImpl::MESSAGE_ID_STRING = {
        "MSG_CONNECT",
        "MSG_SETUP",
        "MSG_SETUP_SUCCESS",
        "MSG_SETUP_FAILED",
        "MSG_SETUP_DONE",
        "MSG_DISCONNECT",
        "MSG_CONNECT_TIMEOUT",
        "MSG_PROCESS_TRIGGER_REQ",
        "MSG_STREAM_RECV_ACTION_EVENT_FROM_PEERS",
        "MSG_STREAM_SEND_ACTION_EVENT_TO_PEERS",
        "MSG_PEER_RENDER_READY",
        "MSG_ERROR",
    };
std::atomic<int> CastSessionImpl::idCount_ = rand();

CastSessionImpl::CastSessionImpl(const CastSessionProperty &property, const CastLocalDevice &localDevice)
    : localDevice_(localDevice), property_(property)
{
    CLOGD("In");
}

CastSessionImpl::~CastSessionImpl()
{
    CLOGD("~CastSessionImpl in");
    StopSafty(false);
    ThreadJoin();
    CLOGD("~CastSessionImpl out");
}

void CastSessionImpl::SetServiceCallbackForRelease(const std::function<void(int)> &callback)
{
    std::lock_guard<std::mutex> lock(serviceCallbackMutex_);
    serviceCallback_ = callback;
}

int32_t CastSessionImpl::RegisterListener(sptr<ICastSessionListenerImpl> listener)
{
    CLOGD("Start to register session listener");
    if (!Permission::CheckPidPermission()) {
        return ERR_NO_PERMISSION;
    }
    std::unique_lock<std::mutex> lock(mutex_);
    listeners_[IPCSkeleton::GetCallingPid()] = listener;
    cond_.notify_all();
    return CAST_ENGINE_SUCCESS;
}

int32_t CastSessionImpl::UnregisterListener()
{
    CLOGD("In");
    std::unique_lock<std::mutex> lock(mutex_);
    listeners_.erase(IPCSkeleton::GetCallingPid());
    return CAST_ENGINE_SUCCESS;
}

int32_t CastSessionImpl::AddDevice(const CastRemoteDevice &remoteDevice)
{
    CLOGI("sessionId_ %{public}d", sessionId_);
    CastInnerRemoteDevice remote = {
        .deviceId = remoteDevice.deviceId,
        .deviceName = remoteDevice.deviceName,
        .deviceType = remoteDevice.deviceType,
        .subDeviceType = remoteDevice.subDeviceType,
        .ipAddress = remoteDevice.ipAddress,
        .sessionId = sessionId_,
        .channelType = remoteDevice.channelType,
    };
    if (!AddDevice(remote)) {
        return CAST_ENGINE_ERROR;
    }
    if (!ConnectionManager::GetInstance().ConnectDevice(remote)) {
        return CAST_ENGINE_ERROR;
    }
    return CAST_ENGINE_SUCCESS;
}

bool CastSessionImpl::AddDevice(const CastInnerRemoteDevice &remoteDevice)
{
    if (!Permission::CheckPidPermission()) {
        return false;
    }

    CLOGI("In: session state: %{public}s", SESSION_STATE_STRING[static_cast<int>(sessionState_)].c_str());
    {
        std::unique_lock<std::mutex> lock(mutex_);
        if (listeners_.empty()) {
            int timeout = 5; // unit: s
            sptr<CastSessionImpl> self(this);
            bool result =
                cond_.wait_for(lock, std::chrono::seconds(timeout), [self] { return !self->listeners_.empty(); });
            if (!result) {
                CLOGE("Wait the listener timeout");
                return false;
            }
        }
    }
    if (!CastDeviceDataManager::GetInstance().UpdateDevice(remoteDevice)) {
        return false;
    }

    if (!AddRemoteDevice(CastRemoteDeviceInfo { remoteDevice, DeviceState::DISCONNECTED })) {
        return false;
    }

    SendCastMessage(Message(MessageId::MSG_CONNECT, remoteDevice.deviceId));
    return true;
}

int32_t CastSessionImpl::RemoveDevice(const std::string &deviceId)
{
    CLOGI("In: session state: %{public}s", SESSION_STATE_STRING[static_cast<int>(sessionState_)].c_str());
    if (!Permission::CheckPidPermission()) {
        return ERR_NO_PERMISSION;
    }

    SendCastMessage(Message(MessageId::MSG_DISCONNECT, deviceId));
    return CAST_ENGINE_SUCCESS;
}

bool CastSessionImpl::ReleaseSessionResources(pid_t pid)
{
    {
        std::lock_guard<std::mutex> lock(mutex_);
        listeners_.erase(pid);
        if (!listeners_.empty()) {
            return false;
        }
        for (auto &deviceInfo : remoteDeviceList_) {
            RemoveDevice(deviceInfo.remoteDevice.deviceId);
        }
    }
    Stop();
    return true;
}

int32_t CastSessionImpl::CreateMirrorPlayer(sptr<IMirrorPlayerImpl> &mirrorPlayer)
{
    if (!Permission::CheckPidPermission()) {
        return ERR_NO_PERMISSION;
    }
    return CAST_ENGINE_SUCCESS;
}

bool CastSessionImpl::DestroyMirrorPlayer()
{
    return true;
}

int32_t CastSessionImpl::CreateStreamPlayer(sptr<IStreamPlayerImpl> &streamPlayer)
{
    if (!Permission::CheckPidPermission()) {
        return ERR_NO_PERMISSION;
    }
    auto streamManager = StreamManagerGetter();
    if (!streamManager) {
        std::unique_lock<std::mutex> lock(streamMutex_);
        streamManager_ =
            ICastStreamManager::GetInstance(std::make_shared<CastStreamListenerImpl>(this), property_.endType, property_.protocolType);
        if (!streamManager_) {
            return CAST_ENGINE_ERROR;
        }
        streamManager = streamManager_;
    }
    streamPlayer = streamManager->CreateStreamPlayer([this]() { DestroyStreamPlayer(); });
    return CAST_ENGINE_SUCCESS;
}

bool CastSessionImpl::DestroyStreamPlayer()
{
    std::unique_lock<std::mutex> lock(streamMutex_);
    streamManager_ = nullptr;
    return true;
}

int32_t CastSessionImpl::Release()
{
    if (!Permission::CheckPidPermission()) {
        return ERR_NO_PERMISSION;
    }
    {
        std::lock_guard<std::mutex> lock(mutex_);
        for (auto &deviceInfo : remoteDeviceList_) {
            RemoveDevice(deviceInfo.remoteDevice.deviceId);
        }
    }
    std::lock_guard<std::mutex> lock(serviceCallbackMutex_);
    if (!serviceCallback_) {
        CLOGE("serviceCallback is null");
        return CAST_ENGINE_ERROR;
    }
    serviceCallback_(sessionId_);
    return CAST_ENGINE_SUCCESS;
}

int32_t CastSessionImpl::StartAuth(const AuthInfo &authInfo)
{
    if (!Permission::CheckPidPermission()) {
        return ERR_NO_PERMISSION;
    }
    static_cast<void>(authInfo);
    return CAST_ENGINE_ERROR;
}

int32_t CastSessionImpl::GetSessionId(std::string &sessionId)
{
    if (!Permission::CheckPidPermission()) {
        return ERR_NO_PERMISSION;
    }
    sessionId = std::to_string(sessionId_);
    return CAST_ENGINE_SUCCESS;
}

int32_t CastSessionImpl::Play(const std::string &deviceId)
{
    if (!Permission::CheckPidPermission()) {
        return ERR_NO_PERMISSION;
    }
    CLOGI("Session state: %{public}s", SESSION_STATE_STRING[static_cast<int>(sessionState_)].c_str());
    SendCastMessage(Message(MessageId::MSG_PLAY, deviceId));
    return CAST_ENGINE_SUCCESS;
}

int32_t CastSessionImpl::Pause(const std::string &deviceId)
{
    if (!Permission::CheckPidPermission()) {
        return ERR_NO_PERMISSION;
    }
    CLOGI("Session state: %{public}s", SESSION_STATE_STRING[static_cast<int>(sessionState_)].c_str());
    SendCastMessage(Message(MessageId::MSG_PAUSE, deviceId));
    return CAST_ENGINE_SUCCESS;
}

int32_t CastSessionImpl::GetDeviceState(const std::string &deviceId, DeviceState &deviceState)
{
    if (!Permission::CheckPidPermission()) {
        return ERR_NO_PERMISSION;
    }
    auto deviceInfo = FindRemoteDevice(deviceId);
    DeviceState state = DeviceState::DISCONNECTED;
    if (deviceInfo != nullptr) {
        state = deviceInfo->deviceState;
    }
    CLOGI("device state: %{public}s", DEVICE_STATE_STRING[static_cast<int>(state)].c_str());
    deviceState = state;
    return CAST_ENGINE_SUCCESS;
}

int32_t CastSessionImpl::SetSessionProperty(const CastSessionProperty &property)
{
    if (!Permission::CheckPidPermission()) {
        return ERR_NO_PERMISSION;
    }

    if (property_.protocolType != property.protocolType || property_.endType != property.endType) {
        CLOGE("Wrong protocol type:%d or end type:%d", property.protocolType, property.endType);
        return CAST_ENGINE_ERROR;
    }
    property_ = property;
    return CAST_ENGINE_SUCCESS;
}

int32_t CastSessionImpl::SetSurface(sptr<IBufferProducer> producer)
{
    if (!Permission::CheckPidPermission()) {
        return ERR_NO_PERMISSION;
    }
    CLOGD("SetSurface in");
    return CAST_ENGINE_SUCCESS;
}

void CastSessionImpl::SetLocalDevice(const CastLocalDevice &localDevice)
{
    localDevice_ = localDevice;
}

bool CastSessionImpl::TransferTo(std::shared_ptr<BaseState> state)
{
    if (IsAllowTransferState(state->GetStateId())) {
        CLOGD("Transfer to %{public}s", SESSION_STATE_STRING[static_cast<int>(state->GetStateId())].c_str());
        TransferState(state);
        return true;
    }
    return false;
}

bool CastSessionImpl::Init()
{
    CLOGI("Session state: %{public}s", SESSION_STATE_STRING[static_cast<int>(sessionState_)].c_str());

    srand(static_cast<unsigned int >(time(nullptr)));
    sessionId_ = rand();
    channelManagerListener_ = std::make_shared<ChannelManagerListenerImpl>(this);
    channelManager_ = std::make_shared<ChannelManager>(sessionId_, channelManagerListener_);

    if (property_.protocolType == ProtocolType::DLNA) {
        dlnaListener_ = std::make_shared<DlnaListenerImpl>(this);
        dlnaControl_ = IDlnaController::GetInstance(dlnaListener_);
        dlnaControl_->DmrCreate();
    } else {
        rtspListener_ = std::make_shared<RtspListenerImpl>(this);
        rtspControl_ = IRtspController::GetInstance(rtspListener_, property_.protocolType, property_.endType);
    }

    defaultState_ = std::make_shared<DefaultState>(SessionState::DEFAULT, this, nullptr);
    disconnectedState_ = std::make_shared<DisconnectedState>(SessionState::DISCONNECTED, this, defaultState_);
    connectingState_ = std::make_shared<ConnectingState>(SessionState::CONNECTING, this, defaultState_);
    connectedState_ = std::make_shared<ConnectedState>(SessionState::CONNECTED, this, defaultState_);
    disconnectingState_ = std::make_shared<DisconnectingState>(SessionState::DISCONNECTING, this, defaultState_);
    pausedState_ = std::make_shared<PausedState>(SessionState::PAUSED, this, connectedState_);
    playingState_ = std::make_shared<PlayingState>(SessionState::PLAYING, this, connectedState_);
    streamState_ = std::make_shared<StreamState>(SessionState::STREAM, this, connectedState_);
    TransferTo(streamState_);

    return true;
}

void CastSessionImpl::Stop()
{
    CLOGD("Start to stop session");
    // remove msg connect timeout to prevent waiting too long for the thread to stop
    RemoveMessage(Message(static_cast<int>(MessageId::MSG_CONNECT_TIMEOUT)));
    StopSafty(true);
    ThreadJoin();
    CLOGD("End to stop session");
}

void CastSessionImpl::InitRtspParamInfo(std::shared_ptr<CastRemoteDeviceInfo> remoteDeviceInfo)
{
    rtspParamInfo_.SetVersion(CAST_VERSION);

    if (property_.protocolType == ProtocolType::CAST_PLUS_STREAM) {
        rtspParamInfo_.SetProjectionMode(CastSessionRtsp::ProjectionMode::STREAM);
    }

    const auto &remote = remoteDeviceInfo->remoteDevice;
    DeviceTypeParamInfo param = {
        .localDeviceType = localDevice_.deviceType,
        .localDeviceSubtype = localDevice_.subDeviceType,
        .remoteDeviceType = remote.deviceType,
        .remoteDeviceSubtype = remote.subDeviceType,
    };
    rtspParamInfo_.SetDeviceTypeParamInfo(param);
    rtspParamInfo_.SetFeatureSet(std::set<int> { ParamInfo::FEATURE_SEND_EVENT_CHANGE });
}

std::string CastSessionImpl::GetCurrentRemoteDeviceId()
{
    std::lock_guard<std::mutex> lock(mutex_);
    if (remoteDeviceList_.empty()) {
        CLOGE("failed, list is empty.");
        return "";
    }
    return remoteDeviceList_.front().remoteDevice.deviceId;
}

bool CastSessionImpl::ProcessConnect(const Message &msg)
{
    auto remoteDeviceInfo = FindRemoteDevice(msg.strArg_);
    if (remoteDeviceInfo == nullptr) {
        CLOGE("remote device is null");
        return false;
    }

    InitRtspParamInfo(remoteDeviceInfo);
    if (!rtspControl_->Start(rtspParamInfo_, remoteDeviceInfo->remoteDevice.sessionKey,
        remoteDeviceInfo->remoteDevice.sessionKeyLength)) {
        CLOGE("Rtsp start failed, session state: %{public}s",
            SESSION_STATE_STRING[static_cast<int>(sessionState_)].c_str());
        rtspControl_->Action(ActionType::TEARDOWN);
        return false;
    }

    const auto &remote = remoteDeviceInfo->remoteDevice;
    CLOGD("DeviceName = %s, deviceId = %s, sessionId = %{public}d.", remote.deviceName.c_str(), remote.deviceId.c_str(),
        remoteDeviceInfo->remoteDevice.sessionId);

    auto request = BuildChannelRequest(remote.deviceId, ModuleType::RTSP);
    if (request == nullptr) {
        CLOGE("Rtsp start failed, session state: %{public}s",
            SESSION_STATE_STRING[static_cast<int>(sessionState_)].c_str());
        return false;
    }

    int deviceSessionId = channelManager_->CreateChannel(*request, rtspControl_->GetChannelListener());
    UpdateRemoteDeviceSessionId(remote.deviceId, deviceSessionId);

    CLOGD("Out: deviceName = %s, deviceId = %s, sessionId = %{public}d.", remote.deviceName.c_str(),
        remote.deviceId.c_str(), remoteDeviceInfo->remoteDevice.sessionId);
    return true;
}

bool CastSessionImpl::ProcessSetUp(const Message &msg)
{
    auto deviceInfo = FindRemoteDevice(msg.strArg_);
    if (deviceInfo == nullptr) {
        CLOGE("Remote device is null");
        rtspControl_->Action(ActionType::TEARDOWN);
        return false;
    }

    const auto &remote = deviceInfo->remoteDevice;
    auto channelType = remote.channelType;
    CLOGD("DeviceName = %s, deviceId = %s, sessionId = %{public}d, channelType = %{public}d.",
        remote.deviceName.c_str(), remote.deviceId.c_str(), remote.sessionId, channelType);
    return true;
}

bool CastSessionImpl::ProcessSetUpSuccess(const Message &msg)
{
    return true;
}

bool CastSessionImpl::ProcessDisconnect(const Message &msg)
{
    CLOGD("In");
    rtspControl_->Action(ActionType::TEARDOWN);
    channelManager_->DestroyAllChannels();
    return true;
}

bool CastSessionImpl::ProcessError(const Message &msg)
{
    CLOGD("In");
    bool result = ProcessDisconnect(msg);
    std::lock_guard<std::mutex> lock(mutex_);
    auto &devices = remoteDeviceList_;
    for (auto it = devices.begin(); it != devices.end();) {
        ChangeDeviceStateLocked(DeviceState::DISCONNECTED, it->remoteDevice.deviceId);
        ConnectionManager::GetInstance().DisconnectDevice(it->remoteDevice.deviceId);
        devices.erase(it++);
    }

    return result;
}

bool CastSessionImpl::ProcessStateEvent(MessageId msgId, const Message &msg)
{
    if (stateProcessor_[msgId] == nullptr) {
        CLOGE("%{public}s' processor is null", MESSAGE_ID_STRING[msgId].c_str());
        return false;
    }

    return (this->*stateProcessor_[msgId])(msg);
}

std::shared_ptr<ChannelRequest> CastSessionImpl::BuildChannelRequest(const std::string &remoteDeviceId,
    ModuleType moduleType)
{
    auto deviceInfo = FindRemoteDevice(remoteDeviceId);
    if (deviceInfo == nullptr) {
        CLOGE("Remote device is null");
        return nullptr;
    }

    const auto &remote = deviceInfo->remoteDevice;
    bool isReceiver = true;

    return std::make_shared<ChannelRequest>(moduleType, isReceiver, localDevice_, remote, property_);
}

std::shared_ptr<CastRemoteDeviceInfo> CastSessionImpl::FindRemoteDevice(const std::string &deviceId)
{
    std::lock_guard<std::mutex> lock(mutex_);
    return FindRemoteDeviceLocked(deviceId);
}

std::shared_ptr<CastRemoteDeviceInfo> CastSessionImpl::FindRemoteDeviceLocked(const std::string &deviceId)
{
    for (auto &deviceInfo : remoteDeviceList_) {
        if (deviceInfo.remoteDevice.deviceId == deviceId) {
            return std::make_shared<CastRemoteDeviceInfo>(deviceInfo);
        }
    }
    return nullptr;
}

void CastSessionImpl::UpdateRemoteDeviceStateLocked(const std::string &deviceId, DeviceState state)
{
    for (auto &deviceInfo : remoteDeviceList_) {
        if (deviceInfo.remoteDevice.deviceId == deviceId) {
            deviceInfo.deviceState = state;
            return;
        }
    }
}

void CastSessionImpl::UpdateRemoteDeviceSessionId(const std::string &deviceId, int sessionId)
{
    std::lock_guard<std::mutex> lock(mutex_);
    for (auto &deviceInfo : remoteDeviceList_) {
        if (deviceInfo.remoteDevice.deviceId == deviceId) {
            deviceInfo.remoteDevice.sessionId = sessionId;
            return;
        }
    }
}

void CastSessionImpl::RemoveRemoteDevice(const std::string &deviceId)
{
    std::lock_guard<std::mutex> lock(mutex_);
    for (auto it = remoteDeviceList_.begin(); it != remoteDeviceList_.end(); it++) {
        if (it->remoteDevice.deviceId == deviceId) {
            CLOGI("Start to remove remote device:%s", deviceId.c_str());
            ConnectionManager::GetInstance().DisconnectDevice(deviceId);
            remoteDeviceList_.erase(it);
            return;
        }
    }
}

bool CastSessionImpl::AddRemoteDevice(const CastRemoteDeviceInfo &remoteDeviceInfo)
{
    const auto &remote = remoteDeviceInfo.remoteDevice;
    if (FindRemoteDevice(remote.deviceId) != nullptr) {
        CLOGW("Remote device(%s) has existed", remote.deviceName.c_str());
        return false;
    }
    std::lock_guard<std::mutex> lock(mutex_);
    remoteDeviceList_.push_back(remoteDeviceInfo);
    return true;
}

// Reserved for 1->N scenarios
bool CastSessionImpl::IsAllowTransferState(SessionState desiredState) const
{
    return true;
}

void CastSessionImpl::ChangeDeviceState(DeviceState state, const std::string &deviceId)
{
    std::lock_guard<std::mutex> lock(mutex_);
    ChangeDeviceStateLocked(state, deviceId);
}

void CastSessionImpl::ChangeDeviceStateLocked(DeviceState state, const std::string &deviceId)
{
    auto deviceInfo = FindRemoteDeviceLocked(deviceId);
    if (!deviceInfo) {
        CLOGE("does not exist this device, deviceId = %s.", deviceId.c_str());
        return;
    }

    CLOGD("New state:%{public}s, old state:%{public}s, device id:%s",
        DEVICE_STATE_STRING[static_cast<int>(state)].c_str(),
        DEVICE_STATE_STRING[static_cast<int>(deviceInfo->deviceState)].c_str(), deviceId.c_str());
    if (state == deviceInfo->deviceState) {
        return;
    }

    UpdateRemoteDeviceStateLocked(deviceId, state);

    for (const auto &[pid, listener] : listeners_) {
        listener->OnDeviceState(DeviceStateInfo { state, deviceId });
    }
}

void CastSessionImpl::OnEvent(EventId eventId, const std::string &data)
{
    if (listeners_.empty()) {
        CLOGE("OnEvent failed because listeners_ is empty!");
        return;
    }
    for (const auto &[pid, listener] : listeners_) {
        listener->OnEvent(eventId, data);
    }
}

bool CastSessionImpl::ProcessTriggerReq(const Message &msg)
{
    return false;
}

void CastSessionImpl::ProcessRtspEvent(int moduleId, int event, const std::string &param)
{
    switch (moduleId) {
        case MODULE_ID_CAST_STREAM:
            SendCastMessage(Message(MessageId::MSG_STREAM_RECV_ACTION_EVENT_FROM_PEERS, event, param));
            break;
        default:
            break;
    }
}

void CastSessionImpl::ProcessDlnaEvent(int moduleId, int event, const std::string &param)
{
	CLOGE("START, moduleId = %{public}d, event = %{public}d ", moduleId, event);
    switch (moduleId) {
        case MODULE_ID_CAST_STREAM:
            SendCastMessage(Message(MessageId::MSG_STREAM_RECV_ACTION_EVENT_FROM_PEERS, event, param));
            break;
        default:
            break;
    }
}

bool CastSessionImpl::IsSupportFeature(const std::set<int> &featureSet, int supportFeature)
{
    return !featureSet.empty() && featureSet.find(supportFeature) != featureSet.end();
}

bool CastSessionImpl::IsConnected() const
{
    return sessionState_ == SessionState::PLAYING || sessionState_ == SessionState::PAUSED ||
        sessionState_ == SessionState::CONNECTED || sessionState_ == SessionState::STREAM;
}

bool CastSessionImpl::SendEventChange(int moduleId, int event, const std::string &param)
{
    CLOGI("Module id %{public}d send event %{public}d", moduleId, event);
    if (property_.protocolType == ProtocolType::DLNA) {
		CLOGI("in");
		return dlnaControl_->SendEventChange(moduleId, event, param);
    } else {
        if (!IsConnected() || !rtspControl_) {
            CLOGE("Send event change fail, state is not ready %{public}hhu", sessionState_);
            return false;
        }

        if (!IsSupportFeature(rtspControl_->GetNegotiatedFeatureSet(), ParamInfo::FEATURE_SEND_EVENT_CHANGE)) {
            CLOGE("The feature is not in the feature set.");
            return false;
        }

        return rtspControl_->SendEventChange(moduleId, event, param);
    }
    return true;
}

std::shared_ptr<ICastStreamManager> CastSessionImpl::StreamManagerGetter()
{
    std::lock_guard<std::mutex> lock(streamMutex_);
    return streamManager_;
}

sptr<IMirrorPlayerImpl> CastSessionImpl::MirrorPlayerGetter()
{
    return nullptr;
}

bool CastSessionImpl::IsStreamMode()
{
    std::lock_guard<std::mutex> lock(streamMutex_);

    if (property_.protocolType == ProtocolType::DLNA) {

    } else {
        return rtspParamInfo_.GetProjectionMode() == CastSessionRtsp::ProjectionMode::STREAM;
    }
    return true;
}

bool CastSessionImpl::IsSink()
{
    std::lock_guard<std::mutex> lock(mutex_);
    return property_.endType == EndType::CAST_SINK;
}

int CastSessionImpl::CreateStreamChannel()
{
    CLOGD("in");
    auto request = BuildChannelRequest(GetCurrentRemoteDeviceId(), ModuleType::STREAM);
    if (request == nullptr) {
        CLOGE("build channel request failed");
        return INVALID_PORT;
    }

    const auto streamManager =  StreamManagerGetter();
    if (channelManager_ == nullptr || streamManager == nullptr) {
        CLOGE("channelManager_ or streamManager is null");
        return INVALID_PORT;
    }
    int port = channelManager_->CreateChannel(*request, streamManager->GetChannelListener());
    if (port == INVALID_PORT) {
        CLOGE("create stream channel failed");
        return INVALID_PORT;
    }
    if (property_.endType == EndType::CAST_SOURCE) {
        SendEventChange(MODULE_ID_CAST_STREAM, ICastStreamManager::MODULE_EVENT_ID_STREAM_CHANNEL,
            std::to_string(port));
    }
    return port;
}

void CastSessionImpl::SendCastRenderReadyOption(int isReady)
{
    std::shared_ptr<IRtspController> rtspControl;
    {
        std::lock_guard<std::mutex> lock(mutex_);
        rtspControl = rtspControl_;
        if (!IsConnected() || !rtspControl || property_.endType != EndType::CAST_SINK) {
            CLOGE("Send render ready failed");
            return;
        }
    }
    rtspControl->SendCastRenderReadyOption(isReady);
}

int32_t CastSessionImpl::NotifyEvent(EventId eventId, std::string &jsonParam)
{
    if (!Permission::CheckPidPermission()) {
        return CAST_ENGINE_ERROR;
    }
    return CAST_ENGINE_SUCCESS;
}

void CastSessionImpl::OnEventInner(sptr<CastSessionImpl> session, EventId eventId, const std::string &jsonParam)
{
    std::unique_lock<std::mutex> lock(mutex_);
    for (const auto &[pid, listener] : session->listeners_) {
        listener->OnEvent(eventId, jsonParam);
    }
}
} // namespace CastEngineService
} // namespace CastEngine
} // namespace OHOS
